package com.hex.ds.hdtp.core.app.data.contrast.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.hex.ds.hdtp.core.app.common.enums.BasicListStatus;
import com.hex.ds.hdtp.core.app.common.enums.BasicTaskStatus;
import com.hex.ds.hdtp.core.app.common.enums.CrossDbContrastInfo;
import com.hex.ds.hdtp.core.app.common.enums.CrossDbContrastResultContrastType;
import com.hex.ds.hdtp.core.app.data.contrast.converter.ICrossDbContrastConverter;
import com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastCommonDto;
import com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastDataSqlDto;
import com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastResultDto;
import com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastSampleDataDto;
import com.hex.ds.hdtp.core.app.data.contrast.service.IContrastSampleDataService;
import com.hex.ds.hdtp.core.app.data.contrast.service.ICrossDbContrastCommonService;
import com.hex.ds.hdtp.core.app.data.contrast.service.ICrossDbContrastRunLogService;
import com.hex.ds.hdtp.core.app.data.table.dto.request.TableTransferFieldMapRequestDto;
import com.hex.ds.hdtp.core.app.data.table.dto.response.TableTransferFieldMapResponseDto;
import com.hex.ds.hdtp.core.app.data.table.service.ITableTransferFieldMapService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceConfInfoService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceInfoService;
import com.hex.ds.hdtp.core.inf.common.exception.SourceJDBCException;
import com.hex.ds.hdtp.core.inf.common.exception.SourceSystemException;
import com.hex.ds.hdtp.core.inf.data.contrast.service.IDataInfoService;
import com.hex.ds.hdtp.core.inf.data.contrast.service.impl.DataInfoSelectHandlerContext;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.repository.impl.SourceConfInfoRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

import static com.hex.ds.hdtp.core.app.data.contrast.service.impl.CrossDbContrastService.getThreadLocal;

/**
 * @Package com.hex.ds.hdtp.core.app.data.contrast.service.impl
 * @ClassName ContrastSampleDataService
 * @Description 全字段明细比对
 * @Author gj.xu
 * @Date 2024/3/4 17:56
 * @Version v2.0
 **/
@Service
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class ContrastSampleDataService implements IContrastSampleDataService {

    private final DataInfoSelectHandlerContext dataInfoSelectHandlerContext;

    private final StreamReaderToContrastService streamReaderToContrastService;

    private final ISourceConfInfoService sourceConfInfoService;

    private final ITableTransferFieldMapService tableTransferFieldMapService;
    private final ICrossDbContrastCommonService crossDbContrastCommonService;

    private final SourceConfInfoRepository sourceConfInfoRepository;

    private final ICrossDbContrastRunLogService crossDbContrastRunLogService;

    private final ISourceInfoService sourceInfoService;

    private final ICrossDbContrastConverter converter;

    static HashMap<String, CrossDbContrastResultDto> crossDbContrastResultDtoMap = new HashMap<>();
    static HashMap<String, CountDownLatch> countDownLatchMap = new HashMap<>();
    static HashMap<String, HashMap<Long, CrossDbContrastDataSqlDto>> sqlDtoMap = new HashMap<>();

    static HashMap<String, Boolean> currentFlag = new HashMap<>();

    static HashMap<String, Boolean> currentIsBatch = new HashMap<>();

    static HashMap<String, Integer> currentCount = new HashMap<>();

    private static final String SOURCE_WHERE = "sourceWhereCondition";

    private static final String TARGET_WHERE = "targetWhereCondition";

    private static final String SOURCE_SORT = "sourceSortField";

    private static final String TARGET_SORT = "targetSortField";

    private IDataInfoService getIDataInfoService(String type) {
        return dataInfoSelectHandlerContext.getDataInfoService(type);
    }

    @Value("${data.detail-contrast.async.corePoolSize}")
    int corePoolSize;

    /*
     * @Method: contrastSampleData <br>
     * @Param: [contrastCommon, contrastSampleData] <br>
     * @Return: boolean <br>
     * @Description：全字段明细比对
     * @Author： wz.li<br>
     * @Date： 2023/7/27 11:07 <br>
     * @Version： V2.0.2<br>
     */
    public boolean contrastSampleData(CrossDbContrastCommonDto contrastCommon, CrossDbContrastSampleDataDto contrastSampleData, String runLogPkId) {
        try {
            // ① 全字段对比
            Boolean allFieldFlag = contrastAllField(contrastCommon, contrastSampleData, runLogPkId);
            return allFieldFlag;
        } catch (Exception e) {
            log.error("全字段明细比对异常：{}", e.getMessage(), e);
            return false;
        }
    }

    /**
     * @Method contrastAllField
     * @Param
     * @Return boolean
     * @Description 全字段明细对比
     * @Author Wang zhihao
     * @Date 2023/8/1 16:33
     * @Version V1.0
     */
    private boolean contrastAllField(CrossDbContrastCommonDto contrastCommon, CrossDbContrastSampleDataDto contrastSampleData, String runLogPkId) {
        CrossDbContrastResultDto resultDto = new CrossDbContrastResultDto();
        if ("Y".equals(contrastSampleData.getIsAllField())) {
            //全字段明细比对逻辑
            resultDto = contrastAllField(runLogPkId, contrastCommon, contrastSampleData);
            if (!resultDto.isContrastResult()) {
                crossDbContrastCommonService.addContrastResult(contrastCommon, runLogPkId, CrossDbContrastResultContrastType.IS_ALL_FIELD.getName(),
                        BasicListStatus.FAIL.name(), resultDto.getSampleData());
                log.error("【数据对比】-[全字段明细对比] 对比失败");
            } else {
                log.info("【数据对比】-[全字段明细对比] 对比成功");
                crossDbContrastCommonService.addContrastResult(contrastCommon, runLogPkId, CrossDbContrastResultContrastType.IS_ALL_FIELD.getName(),
                        BasicListStatus.SUCCESS.name(), "全字段明细对比成功");
            }
        }
        if (crossDbContrastResultDtoMap.get(getThreadLocal()) != null)
            crossDbContrastResultDtoMap.remove(getThreadLocal());
        return resultDto.isContrastResult();
    }

    /**
     * @param contrastCommon
     * @Method contrastAllField
     * @Param
     * @Return java.lang.Boolean
     * @Description 全字段明细对比
     * @Author Wang zhihao
     * @Date 2023/8/18 9:24
     * @Version V1.0
     */
    public CrossDbContrastResultDto contrastAllField(String runLogPkId, CrossDbContrastCommonDto contrastCommon, CrossDbContrastSampleDataDto contrastSampleData) {
        try {
            //初始化
            crossDbContrastResultDtoMap.put(getThreadLocal(), new CrossDbContrastResultDto());
            currentFlag.put(getThreadLocal(), Boolean.TRUE);
            currentIsBatch.put(getThreadLocal(), Boolean.TRUE);
            currentCount.put(getThreadLocal(), corePoolSize);
            //1.数量标度
            Long multiple = 10000L;
            //2.查询比对表的数据量
            List<SourceConfInfoPo> sourceConfInfoPoList = sourceConfInfoRepository.queryListBySourceCode(contrastCommon.getSourceCode());
            List<SourceConfInfoPo> tartgetConfInfoPoList = sourceConfInfoRepository.queryListBySourceCode(contrastCommon.getContrastCode());
            CrossDbContrastDataSqlDto dataSqlDto = crossDbContrastCommonService.buildPartitionWhereSql(sourceConfInfoPoList.get(0).getSourceType(), tartgetConfInfoPoList.get(0).getSourceType(), contrastCommon);
            String sourceWhereCondition = "";
            if (!StrUtil.isEmpty(contrastSampleData.getWhereCondition())) {
                JSONObject jsonObject = JSONObject.parseObject(contrastSampleData.getWhereCondition());
                sourceWhereCondition = jsonObject.getString(SOURCE_WHERE).replace(";", "");
            }
            sourceWhereCondition = getWhereSql(dataSqlDto.getSourcePartitionWhereSql(), sourceWhereCondition);
            Long longResult = getIDataInfoService(sourceConfInfoPoList.get(0).getSourceType())
                    .selectDataNum(contrastCommon.getSourceCode(), contrastCommon.getSourceSchema(), contrastCommon.getSourceTable(), sourceWhereCondition, "");
            if (crossDbContrastRunLogService.queryStatusByPkId(runLogPkId) != null
                    && crossDbContrastRunLogService.queryStatusByPkId(runLogPkId).equalsIgnoreCase(BasicTaskStatus.STOPPED.name()))
                return null;

            // 处理不同数据量范围时数据比对
            return handleDiffDataScop(runLogPkId, contrastCommon, contrastSampleData, longResult, multiple);

        } catch (Exception e) {
            log.error("全字段明细对比发生异常: {}", e.getMessage(), e);
            return new CrossDbContrastResultDto().setContrastResult(false).setSampleData(e.getMessage());
        } finally {
            if (sqlDtoMap.get(getThreadLocal()) != null && !sqlDtoMap.get(getThreadLocal()).isEmpty())
                sqlDtoMap.get(getThreadLocal()).clear();
            if (countDownLatchMap.get(getThreadLocal()) != null) countDownLatchMap.remove(getThreadLocal());
        }
    }

    /**
     * @Method handleDiffDataScop
     * @Param runLogPkId
     * @Param contrastCommon
     * @Param contrastSampleData
     * @Param longResult
     * @Param multiple
     * @Return com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastResultDto
     * @Description 处理不同数据量范围时数据比对
     * @Author gj.xu
     * @Date 2024/2/22 16:20
     * @Version V2.0
     */
    private CrossDbContrastResultDto handleDiffDataScop(String runLogPkId, CrossDbContrastCommonDto contrastCommon, CrossDbContrastSampleDataDto contrastSampleData, Long longResult, Long multiple) {
        if (longResult.compareTo((contrastSampleData.getDataExtractScope() * multiple)) <= 0) {
            // 3.1 数据量小于 数据量范围
            if ("CUSTOM".equals(contrastSampleData.getLessExtractScopeType())) {
                // 3.1.1 分批次读取比对
                // 计算 需要分多少批 和 每批次数据量
                long count = (longResult % (contrastSampleData.getLessExtractScopeCustom() * multiple)) == 0 ? (longResult / (contrastSampleData.getLessExtractScopeCustom() * multiple)) : ((longResult / (contrastSampleData.getLessExtractScopeCustom() * multiple)) + 1);
                final long extractScope = contrastSampleData.getLessExtractScopeCustom() * multiple;
                return contrastBatchReadData(runLogPkId, contrastCommon, contrastSampleData, count, extractScope);
            } else {
                // 3.1.2 全量读取比对
                return contrastAllData(runLogPkId, contrastCommon, contrastSampleData);
            }
        } else {
            // 3.2 数据量大于 数据量范围
            if ("CUSTOM".equals(contrastSampleData.getGreaterExtractScopeType())) {
                // 3.2.1 分批次读取比对
                // 计算 需要分多少批 和 每批次数据量
                long count = (longResult % (contrastSampleData.getGreaterExtractScopeCustom() * multiple)) == 0 ? (longResult / (contrastSampleData.getGreaterExtractScopeCustom() * multiple)) : ((longResult / (contrastSampleData.getGreaterExtractScopeCustom() * multiple)) + 1);
                final long extractScope = contrastSampleData.getGreaterExtractScopeCustom() * multiple;
                return contrastBatchReadData(runLogPkId, contrastCommon, contrastSampleData, count, extractScope);
            } else {
                // 3.2.2 全量读取比对
                return contrastAllData(runLogPkId, contrastCommon, contrastSampleData);
            }
        }
    }

    /**
     * @Method contrastBatchReadData
     * @Param runLogPkId
     * @Param contrastCommon
     * @Param contrastSampleData
     * @Param count
     * @Param extractScope
     * @Return com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastResultDto
     * @Description 根据配置分批读取读取数据进行比对
     * @Author gj.xu
     * @Date 2024/2/22 14:59
     * @Version V2.0
     */
    private CrossDbContrastResultDto contrastBatchReadData(String runLogPkId, CrossDbContrastCommonDto contrastCommon, CrossDbContrastSampleDataDto contrastSampleData, long count, long extractScope) {
        try {
            countDownLatchMap.put(getThreadLocal(), new CountDownLatch((int) count));
            for (long i = 0; i < count; i++) {
                if (crossDbContrastRunLogService.queryStatusByPkId(runLogPkId) != null
                        && crossDbContrastRunLogService.queryStatusByPkId(runLogPkId).equalsIgnoreCase(BasicTaskStatus.STOPPED.name()))
                    return null;
                if (!crossDbContrastResultDtoMap.get(getThreadLocal()).isContrastResult() && crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList().size() >= contrastSampleData.getSampleDataCount()) {
                    integrationMistakeResult(contrastCommon, contrastSampleData);
                    return crossDbContrastResultDtoMap.get(getThreadLocal());
                }
                // 构建分批读取比对sql
                CrossDbContrastDataSqlDto sql = querySqlLimit(contrastCommon, contrastSampleData, extractScope * i, extractScope);
                HashMap<Long, CrossDbContrastDataSqlDto> sqlDtoHashMap = new HashMap<>();
                if (sqlDtoMap.get(getThreadLocal()) != null)
                    sqlDtoHashMap = sqlDtoMap.get(getThreadLocal());
                sqlDtoHashMap.put(i, sql);
                sqlDtoMap.put(getThreadLocal(), sqlDtoHashMap);
                // 比对逻辑
                streamReaderToContrastService.sampleContrast(contrastCommon, contrastSampleData, i, getThreadLocal(), runLogPkId);
            }
            countDownLatchMap.get(getThreadLocal()).await();
            integrationMistakeResult(contrastCommon, contrastSampleData);
            return crossDbContrastResultDtoMap.get(getThreadLocal());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new SourceJDBCException("根据配置分批读取读取数据进行比对失败", e);
        } finally {
            clearCurrentContrast();
        }
    }

    /**
     * @Method clearCurrentContrast
     * @Param
     * @Return void
     * @Description 清理异步比对信息
     * @Author gj.xu
     * @Date 2024/3/4 21:33
     * @Version V2.0
     */
    private void clearCurrentContrast() {
        if (sqlDtoMap.get(getThreadLocal()) != null && !sqlDtoMap.get(getThreadLocal()).isEmpty())
            sqlDtoMap.get(getThreadLocal()).clear();
        if (countDownLatchMap.get(getThreadLocal()) != null) countDownLatchMap.remove(getThreadLocal());
        if (currentIsBatch.get(getThreadLocal()) != null) currentIsBatch.remove(getThreadLocal());
        if (currentFlag.get(getThreadLocal()) != null) currentFlag.remove(getThreadLocal());
        if (currentCount.get(getThreadLocal()) != null) currentCount.remove(getThreadLocal());
    }

    /**
     * @Method contrastAllData
     * @Param runLogPkId
     * @Param contrastCommon
     * @Param contrastSampleData
     * @Return com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastResultDto
     * @Description 全量读取数据进行比对
     * @Author gj.xu
     * @Date 2024/2/22 14:38
     * @Version V2.0
     */
    private CrossDbContrastResultDto contrastAllData(String runLogPkId, CrossDbContrastCommonDto contrastCommon, CrossDbContrastSampleDataDto contrastSampleData) {
        currentIsBatch.put(getThreadLocal(), Boolean.FALSE);
        //构建全量比对sql
        CrossDbContrastDataSqlDto sql = querySql(contrastCommon, contrastSampleData);
        //比对逻辑
        try {
            return streamReaderToContrastService.streamContrast(contrastCommon, sql, contrastSampleData, getThreadLocal(), runLogPkId);
        } catch (Exception e) {
            throw new SourceSystemException("全量读取数据进行比对异常", e);
        }
    }

    /*
     * @param contrastInfo:
     * @param crossDbContrastSampleDataDto:
     * @return void
     * @author Liang Fuhao
     * @description 整合错误信息
     * @date 2024/1/29 17:46
     */
    public void integrationMistakeResult(CrossDbContrastCommonDto contrastInfo, CrossDbContrastSampleDataDto crossDbContrastSampleDataDto) {
        if (crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList() != null && crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList().size() != 0)
            crossDbContrastResultDtoMap.get(getThreadLocal()).setContrastResult(false);
        if (currentFlag.get(getThreadLocal()) || crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList().size() < crossDbContrastSampleDataDto.getSampleDataCount())
            return;
        List<CrossDbContrastResultDto.SampleData> sampleData = new ArrayList<>();
        if (crossDbContrastSampleDataDto.getSortField() == null || crossDbContrastSampleDataDto.getSortField().isEmpty()) {
            List<SourceConfInfoPo> sourceConfInfoPoList = sourceConfInfoService.queryListBySourceCode(contrastInfo.getSourceCode());
            List<String> sourceDataInfo = getIDataInfoService(sourceConfInfoPoList.get(0).getSourceType())
                    .selectFieldName(contrastInfo.getSourceCode(), contrastInfo.getSourceSchema(), contrastInfo.getSourceTable());
            String sourceSortField = StringUtils.join(sourceDataInfo, ",");
            sortByOrderFields(crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList(), sourceSortField.split(","));
//            crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList().sort(Comparator.comparing(CrossDbContrastResultDto.SampleData::toString));
        } else {
            List<SourceConfInfoPo> sourceConfInfoPoList = sourceConfInfoService.queryListBySourceCode(contrastInfo.getSourceCode());
            List<SourceConfInfoPo> targetConfInfoPoList = sourceConfInfoService.queryListBySourceCode(contrastInfo.getContrastCode());
            List<String> sourceDataInfo = getIDataInfoService(sourceConfInfoPoList.get(0).getSourceType())
                    .selectFieldName(contrastInfo.getSourceCode(), contrastInfo.getSourceSchema(), contrastInfo.getSourceTable());
            List<String> targetDataInfo = getIDataInfoService(targetConfInfoPoList.get(0).getSourceType())
                    .selectFieldName(contrastInfo.getContrastCode(), contrastInfo.getContrastSchema(), contrastInfo.getContrastTable());
            String sourceSortField = StringUtils.join(sourceDataInfo, ",");
            String targetSortField = StringUtils.join(targetDataInfo, ",");
            if (!StringUtils.isEmpty(crossDbContrastSampleDataDto.getSortField())) {
                JSONObject jsonObject = JSONObject.parseObject(crossDbContrastSampleDataDto.getSortField());
                String getSourceSortField = jsonObject.getString(SOURCE_SORT).replace(";", "");
                String getTargetSortField = jsonObject.getString(TARGET_SORT).replace(";", "");
                sourceSortField = StrUtil.isNotBlank(getSourceSortField) ? getSourceSortField : sourceSortField;
                targetSortField = StrUtil.isNotBlank(getTargetSortField) ? getTargetSortField : targetSortField;
            }
            sortByOrderFields(crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList(), targetSortField.split(","));
            sortByOrderFields(crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList(), sourceSortField.split(","));
        }
        fillErrorSampleData(crossDbContrastSampleDataDto, sampleData);
        crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList().clear();
        crossDbContrastResultDtoMap.get(getThreadLocal()).setSampleDataList(sampleData);
        crossDbContrastResultDtoMap.get(getThreadLocal()).setSampleData(JSONObject.toJSONString(sampleData, SerializerFeature.WriteMapNullValue));
    }

    /**
     * @Method fillErrorSampleData
     * @Param crossDbContrastSampleDataDto
     * @Param sampleData
     * @Return void
     * @Description 填充抽取的错误数据
     * @Author gj.xu
     * @Date 2024/2/22 16:09
     * @Version V2.0
     */
    private void fillErrorSampleData(CrossDbContrastSampleDataDto crossDbContrastSampleDataDto, List<CrossDbContrastResultDto.SampleData> sampleData) {
        for (int i = 0; i < crossDbContrastSampleDataDto.getSampleDataCount(); i++) {
            ArrayList<String> fields = new ArrayList<>();
            Map<String, Object> sourceMap = crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList().get(i).getSource();
            Map<String, Object> targetMap = crossDbContrastResultDtoMap.get(getThreadLocal()).getSampleDataList().get(i).getTarget();
            Iterator<String> iterator = sourceMap.keySet().iterator();
            while (iterator.hasNext()) {
                String next = iterator.next();
                Object sourceData = sourceMap.get(next);
                Object targetData = targetMap.get(next);
                if (sourceData != null && !sourceData.equals(targetData)) {
                    fields.add(next);
                } else if (sourceData == null && targetData != null) {
                    fields.add(next);
                }
            }
            sampleData.add(new CrossDbContrastResultDto.SampleData().setFields(fields).setSource(sourceMap).setTarget(targetMap));
        }
    }

    private static void sortByOrderFields(List<CrossDbContrastResultDto.SampleData> list, String[] keysToSort) {
        //根据order by字段排序的比较器(升序)
        Comparator<CrossDbContrastResultDto.SampleData> comparator = (data1, data2) -> {
            for (String key : keysToSort) {
                Comparable<Object> value1 = (Comparable<Object>) data1.getSource().get(key);
                Comparable<Object> value2 = (Comparable<Object>) data2.getSource().get(key);
                if (value1 != null && value2 != null) {
                    int result = value1.compareTo(value2);
                    if (result != 0) {
                        return result;
                    }
                }
            }
            return 0;
        };
        Collections.sort(list, comparator);
    }


    /**
     * @Method querySqlLimit
     * @Param
     * @Param contrastInfo
     * @Return com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastDataSqlDto
     * @Description 指定数量 查询sql
     * @Author xin.zhao
     * @Date 2023/10/9 17:30
     * @Version V1.0
     */
    public CrossDbContrastDataSqlDto querySqlLimit(CrossDbContrastCommonDto contrastInfo, CrossDbContrastSampleDataDto contrastSampleData, Long offset, Long extractScope) {
        CrossDbContrastDataSqlDto dataSqlDto = new CrossDbContrastDataSqlDto();
        try {
            List<SourceConfInfoPo> sourceConfInfoPoList = sourceConfInfoService.queryListBySourceCode(contrastInfo.getSourceCode());
            List<SourceConfInfoPo> contrastConfInfoPoList = sourceConfInfoService.queryListBySourceCode(contrastInfo.getContrastCode());
            // 获取场景化下分区字段处理后的字段名
            List<String> sourceDataInfo = getIDataInfoService(sourceConfInfoPoList.get(0).getSourceType())
                    .selectFieldName(contrastInfo.getSourceCode(), contrastInfo.getSourceSchema(), contrastInfo.getSourceTable());
            List<String> targetDataInfo = getIDataInfoService(contrastConfInfoPoList.get(0).getSourceType())
                    .selectFieldName(contrastInfo.getContrastCode(), contrastInfo.getContrastSchema(), contrastInfo.getContrastTable());
            // 获取源与对比目标类型
            String sourceType = sourceConfInfoPoList.get(0).getSourceType();
            String contrastType = contrastConfInfoPoList.get(0).getSourceType();

            // 如果比对双方不为相同类型且源表是IMPALA或HIVE(大数据仓库)类型，若对比表中不存在源表的分区同名字段 则IMPALA类型的表去掉相应字段
            if (!sourceType.equals(contrastType) && (sourceType.equals(CrossDbContrastInfo.IMPALA.getName()) || sourceType.equals(CrossDbContrastInfo.HIVE.getName()))) {
                List<String> partitionField = getIDataInfoService(sourceType).selectPartitionField(contrastInfo.getSourceCode(), contrastInfo.getSourceSchema(), contrastInfo.getSourceTable());
                dealFieldName(sourceDataInfo, targetDataInfo, partitionField);
            } else if (!sourceType.equals(contrastType) && (contrastType.equals(CrossDbContrastInfo.IMPALA.getName())
                    || contrastType.equals(CrossDbContrastInfo.GAUSS.getName())
                    || contrastType.equals(CrossDbContrastInfo.HIVE.getName()))) {
                List<String> partitionField = getIDataInfoService(contrastType).selectPartitionField(contrastInfo.getContrastCode(), contrastInfo.getContrastSchema(), contrastInfo.getContrastTable());
                dealFieldName(targetDataInfo, sourceDataInfo, partitionField);
            }
            sourceDataInfo = sourceDataInfo.stream().sorted(Comparator.comparing(String::toString)).collect(Collectors.toList());
            targetDataInfo = sourceDataInfo.stream().sorted(Comparator.comparing(String::toString)).collect(Collectors.toList());
            return getAllFieldSqlLimit(sourceType, contrastType, sourceDataInfo, targetDataInfo, contrastInfo, contrastSampleData, extractScope, offset);
        } catch (Exception e) {
            log.error("明细对比发生异常: {}", e.getMessage(), e);
        }
        return dataSqlDto;
    }

    /**
     * @Method getAllFieldSqlLimit
     * @Param
     * @Param sourceType
     * @Param contrastType
     * @Param sourceDataInfo
     * @Param targetDataInfo
     * @Param contrastInfo
     * @Param extractScope
     * @Param offset
     * @Return com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastDataSqlDto
     * @Description 分批读取比对sql
     * @Author xin.zhao
     * @Date 2023/10/10 10:36
     * @Version V1.0
     */
    public CrossDbContrastDataSqlDto getAllFieldSqlLimit(String sourceType, String contrastType,
                                                         List<String> sourceDataInfo, List<String> targetDataInfo, CrossDbContrastCommonDto contrastInfo, CrossDbContrastSampleDataDto contrastSampleData, Long extractScope, Long offset) throws SQLException {
        CrossDbContrastDataSqlDto dataSqlDto = crossDbContrastCommonService.buildPartitionWhereSql(sourceType, contrastType, contrastInfo);
        String sourceWhereCondition = "";
        String targetWhereCondition = "";
        if (!StrUtil.isEmpty(contrastSampleData.getWhereCondition())) {
            JSONObject jsonObject = JSONObject.parseObject(contrastSampleData.getWhereCondition());
            sourceWhereCondition = jsonObject.getString(SOURCE_WHERE).replace(";", "");
            targetWhereCondition = jsonObject.getString(TARGET_WHERE).replace(";", "");
        }
        String sourceSortField = StringUtils.join(sourceDataInfo, ",");
        String targetSortField = StringUtils.join(targetDataInfo, ",");
        if (!StringUtils.isEmpty(contrastSampleData.getSortField())) {
            JSONObject jsonObject = JSONObject.parseObject(contrastSampleData.getSortField());
            sourceSortField = StrUtil.isNotBlank(jsonObject.getString(SOURCE_SORT).replace(";", "")) ? jsonObject.getString(SOURCE_SORT).replace(";", "") : sourceSortField;
            targetSortField = StrUtil.isNotBlank(jsonObject.getString(TARGET_SORT).replace(";", "")) ? jsonObject.getString(TARGET_SORT).replace(";", "") : targetSortField;
        }
        // 获取查询数据信息的Sql语句
        String sourceAllFieldSql = getIDataInfoService(sourceType).selectAllFieldSqlLimit(
                sourceType,
                StringUtils.join(sourceDataInfo, ","),
                contrastInfo.getSourceSchema(),
                contrastInfo.getSourceTable(),
                getWhereSql(dataSqlDto.getSourcePartitionWhereSql(), sourceWhereCondition),
                extractScope,
                offset, sourceSortField);
        String targetAllFieldSql = getIDataInfoService(contrastType).selectAllFieldSqlLimit(
                contrastType,
                StringUtils.join(targetDataInfo, ","),
                contrastInfo.getContrastSchema(),
                contrastInfo.getContrastTable(),
                getWhereSql(dataSqlDto.getTargetPartitionWhereSql(), targetWhereCondition),
                extractScope,
                offset, targetSortField);
        return dataSqlDto.setSourceQuerySql(sourceAllFieldSql).setTargetQuerySql(targetAllFieldSql);
    }

    /**
     * @Method querySql <br>
     * @Param
     * @Param contrastInfo <br>
     * @Return com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastDataSqlDto <br>
     * @Description 查询sql<br>
     * @Author jian.qiao<br>
     * @Date 2023/9/26 16:41 <br>
     * @Version V1.0<br>
     */
    public CrossDbContrastDataSqlDto querySql(CrossDbContrastCommonDto contrastInfo, CrossDbContrastSampleDataDto crossDbContrastSampleDataDto) {
        CrossDbContrastDataSqlDto dataSqlDto = new CrossDbContrastDataSqlDto();
        try {
            //查询源和目标库表信息
            List<SourceConfInfoPo> sourceConfInfoPoList = sourceConfInfoService.queryListBySourceCode(contrastInfo.getSourceCode());
            List<SourceConfInfoPo> contrastConfInfoPoList = sourceConfInfoService.queryListBySourceCode(contrastInfo.getContrastCode());
            // 获取场景化下分区字段处理后的字段名
            List<String> sourceDataInfo = getIDataInfoService(sourceConfInfoPoList.get(0).getSourceType())
                    .selectFieldName(contrastInfo.getSourceCode(), contrastInfo.getSourceSchema(), contrastInfo.getSourceTable());
            List<String> targetDataInfo = getIDataInfoService(contrastConfInfoPoList.get(0).getSourceType())
                    .selectFieldName(contrastInfo.getContrastCode(), contrastInfo.getContrastSchema(), contrastInfo.getContrastTable());
            //字段名映射
            List<TableTransferFieldMapResponseDto> tableTransferFieldMaps = new ArrayList<>();
            if (contrastInfo.getTableTransferPkId() != null) {
                //如果经历过表迁移，那么就从表迁移里获取字段映射
                TableTransferFieldMapRequestDto tableTransferFieldMapRequestDto = new TableTransferFieldMapRequestDto();
                tableTransferFieldMapRequestDto.setListPkId(contrastInfo.getTableTransferPkId());
                tableTransferFieldMaps = tableTransferFieldMapService.queryList(tableTransferFieldMapRequestDto);
            } else {
                //否则就连接表获取字段映射
                tableTransferFieldMaps = sourceInfoService.getFieldMaps(converter.toBasicTransferInfo(contrastInfo));
            }
            //获取字段映射
            Map<String, String> fieldsMap = tableTransferFieldMaps.stream()
                    .collect(Collectors.toMap(TableTransferFieldMapResponseDto::getSourceFieldName, TableTransferFieldMapResponseDto::getTargetFieldName));
            //排除字段逻辑
            if (!CollUtil.isEmpty(crossDbContrastSampleDataDto.getExcludeField())) {
                List<String> excludeFields = crossDbContrastSampleDataDto.getExcludeField().stream().map(f -> f.toUpperCase()).collect(Collectors.toList());
                List<String> excludeTargetFields = crossDbContrastSampleDataDto.getExcludeField().stream().map(f -> fieldsMap.get(f).toUpperCase()).collect(Collectors.toList());
                sourceDataInfo = sourceDataInfo.stream().filter(f -> !excludeFields.contains(f)).collect(Collectors.toList());
                targetDataInfo = targetDataInfo.stream().filter(f -> !excludeTargetFields.contains(f)).collect(Collectors.toList());
            }
            // 获取源与对比目标类型
            String sourceType = sourceConfInfoPoList.get(0).getSourceType();
            String contrastType = contrastConfInfoPoList.get(0).getSourceType();
            // 如果比对双方不为相同类型且源表是IMPALA或HIVE(大数据仓库)类型，若对比表中不存在源表的分区同名字段 则IMPALA类型的表去掉相应字段
            if (!sourceType.equals(contrastType) && (sourceType.equals(CrossDbContrastInfo.IMPALA.getName()) || sourceType.equals(CrossDbContrastInfo.HIVE.getName()))) {
                List<String> partitionField = getIDataInfoService(sourceType).selectPartitionField(contrastInfo.getSourceCode(), contrastInfo.getSourceSchema(), contrastInfo.getSourceTable());
                dealFieldName(sourceDataInfo, targetDataInfo, partitionField);
            } else if (!sourceType.equals(contrastType) && (contrastType.equals(CrossDbContrastInfo.IMPALA.getName())
                    || contrastType.equals(CrossDbContrastInfo.GAUSS.getName())
                    || contrastType.equals(CrossDbContrastInfo.HIVE.getName()))) {
                List<String> partitionField = getIDataInfoService(contrastType).selectPartitionField(contrastInfo.getContrastCode(), contrastInfo.getContrastSchema(), contrastInfo.getContrastTable());
                dealFieldName(targetDataInfo, sourceDataInfo, partitionField);
            }
            //获取源和目标字段信息
            sourceDataInfo = sourceDataInfo.stream().sorted(Comparator.comparing(String::toString)).collect(Collectors.toList());
            targetDataInfo = sourceDataInfo.stream().sorted(Comparator.comparing(String::toString)).collect(Collectors.toList());
            //构建sql
            return getAllFieldSql(sourceType, contrastType, sourceDataInfo, targetDataInfo, contrastInfo, crossDbContrastSampleDataDto);
        } catch (Exception e) {
            log.error("明细对比发生异常: {}", e.getMessage(), e);
        }
        return dataSqlDto;
    }

    /**
     * @Method getAllFieldSql <br>
     * @Param sourceType
     * @Param contrastType
     * @Param sourceDataInfo
     * @Param targetDataInfo
     * @Param contrastInfo <br>
     * @Return com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastDataSqlDto <br>
     * @Description 查询sql<br>
     * @Author jian.qiao<br>
     * @Date 2023/10/8 16:05 <br>
     * @Version V1.0<br>
     */
    public CrossDbContrastDataSqlDto getAllFieldSql(String sourceType, String contrastType,
                                                    List<String> sourceDataInfo, List<String> targetDataInfo, CrossDbContrastCommonDto contrastInfo, CrossDbContrastSampleDataDto crossDbContrastSampleDataDto) throws SQLException {
        CrossDbContrastDataSqlDto dataSqlDto = crossDbContrastCommonService.buildPartitionWhereSql(sourceType, contrastType, contrastInfo);
        String sourceWhereCondition = "";
        String targetWhereCondition = "";
        if (!StrUtil.isEmpty(crossDbContrastSampleDataDto.getWhereCondition())) {
            JSONObject jsonObject = JSONObject.parseObject(crossDbContrastSampleDataDto.getWhereCondition());
            sourceWhereCondition = jsonObject.getString(SOURCE_WHERE).replace(";", "");
            targetWhereCondition = jsonObject.getString(TARGET_WHERE).replace(";", "");
        }
        String sourceSortField = StringUtils.join(sourceDataInfo, ",");
        String targetSortField = StringUtils.join(targetDataInfo, ",");
        if (!StringUtils.isEmpty(crossDbContrastSampleDataDto.getSortField())) {
            JSONObject jsonObject = JSONObject.parseObject(crossDbContrastSampleDataDto.getSortField());
            sourceSortField = StrUtil.isNotBlank(jsonObject.getString(SOURCE_SORT).replace(";", "")) ? jsonObject.getString(SOURCE_SORT).replace(";", "") : sourceSortField;
            targetSortField = StrUtil.isNotBlank(jsonObject.getString(TARGET_SORT).replace(";", "")) ? jsonObject.getString(TARGET_SORT).replace(";", "") : targetSortField;
        }
        // 获取查询数据信息的Sql语句
        String sourceAllFieldSql = getIDataInfoService(sourceType).selectAllFieldSql(
                sourceType,
                StringUtils.join(sourceDataInfo, ","),
                contrastInfo.getSourceSchema(),
                contrastInfo.getSourceTable(),
                getWhereSql(dataSqlDto.getSourcePartitionWhereSql(), sourceWhereCondition), sourceSortField);
        String targetAllFieldSql = getIDataInfoService(contrastType).selectAllFieldSql(
                contrastType,
                StringUtils.join(targetDataInfo, ","),
                contrastInfo.getContrastSchema(),
                contrastInfo.getContrastTable(),
                getWhereSql(dataSqlDto.getTargetPartitionWhereSql(), targetWhereCondition), targetSortField);
        return dataSqlDto.setSourceQuerySql(sourceAllFieldSql).setTargetQuerySql(targetAllFieldSql);
    }

    /**
     * @param dealFieldList
     * @param referFieldList
     * @Method dealFieldName
     * @Param
     * @Return java.util.List<java.lang.String>
     * @Description 处理因分区字段场景导致的字段问题
     * @Author Wang zhihao
     * @Date 2023/8/22 15:27
     * @Version V1.0
     */
    private List<String> dealFieldName(List<String> dealFieldList, List<String> referFieldList, List<String> partitionFieldList) {
        for (String partitionField : partitionFieldList) {
            if (!referFieldList.contains(partitionField)) {
                dealFieldList.remove(partitionField);
            }
            for (String referField : referFieldList) {
                if (partitionField.equalsIgnoreCase(referField.concat(CrossDbContrastInfo.SUFFIX.getName()))) {
                    dealFieldList.remove(partitionField);
                }
            }
        }
        return dealFieldList;
    }

    /**
     * @Method getWhereSql
     * @Param
     * @Param partitionWhereSql
     * @Param whereCondition
     * @Return java.lang.String
     * @Description sqlWHERE条件拼接
     * @Author xin.zhao
     * @Date 2023/10/26 20:06
     * @Version V1.0
     */
    private String getWhereSql(String partitionWhereSql, String whereCondition) {
        if (com.hex.bigdata.hbda.utils.StringUtils.isNotBlank(partitionWhereSql)) {
            if (com.hex.bigdata.hbda.utils.StringUtils.isNotBlank(whereCondition)) {
                return partitionWhereSql + " and " + whereCondition;
            } else {
                return partitionWhereSql;
            }
        } else {
            if (com.hex.bigdata.hbda.utils.StringUtils.isNotBlank(whereCondition)) {
                return "where " + whereCondition;
            } else {
                return "";
            }
        }
    }
}
